package com.xiaomaoguai.fcp.pre.kepler.router.rpc.server;

import com.xiaomaoguai.fcp.pre.kepler.router.rpc.options.ServerOptions;
import com.xiaomaoguai.fcp.pre.kepler.router.rpc.server.channelInitializer.NettyChannelInitializer;
import io.netty.bootstrap.Bootstrap;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.WriteBufferWaterMark;
import io.netty.channel.epoll.Epoll;
import io.netty.channel.epoll.EpollEventLoopGroup;
import io.netty.channel.epoll.EpollServerSocketChannel;
import io.netty.channel.kqueue.KQueue;
import io.netty.channel.kqueue.KQueueEventLoopGroup;
import io.netty.channel.kqueue.KQueueServerSocketChannel;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.util.concurrent.DefaultEventExecutorGroup;
import io.netty.util.concurrent.DefaultThreadFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.ThreadFactory;

/**
 * NettyServer
 *
 * @author DH
 */
public class NettyServer {

    protected final Integer port;

    private final Logger log = LoggerFactory.getLogger(getClass());

    private final Integer nBosses;

    private final Integer nWorkers;

    private Integer connections;

    private ServerBootstrap bootstrap;

    private EventLoopGroup boss;

    private EventLoopGroup worker;

    private DefaultEventExecutorGroup Processor;

    private volatile Channel channel;

    private NettyChannelInitializer channelInitializer;

    public NettyServer(Integer port, NettyChannelInitializer channelInitializer) {
        this.port = port;
        this.nBosses = Math.max(2, ServerOptions.ACCPET_THREADS);
        this.nWorkers = Math.max(4, ServerOptions.IO_THREADS);
        this.channelInitializer = channelInitializer;
    }

    protected void start() throws InterruptedException {
        boss = initEventLoopGroup(nBosses);
        worker = initEventLoopGroup(nWorkers);
        bootstrap = initChannel();
        bootstrap.option(ChannelOption.SO_REUSEADDR, true)
                .option(ChannelOption.SO_RCVBUF, 256 * 1024)
                .option(ChannelOption.SO_TIMEOUT, ServerOptions.TIMEOUT)
                .option(ChannelOption.SO_BACKLOG, connections == null ? ServerOptions.CONNECTIONS : connections)
                .option(ChannelOption.TCP_NODELAY, ServerOptions.TCP_NODELAY)
                .childOption(ChannelOption.SO_KEEPALIVE, true)
                .childOption(ChannelOption.SO_REUSEADDR, true)
                .childOption(ChannelOption.WRITE_BUFFER_WATER_MARK, new WriteBufferWaterMark(1024 * 1024, 2048 * 1024));
        bootstrap.childHandler(channelInitializer);
        channel = bootstrap.bind(port).sync().channel();
        // Netty启动环境相关信息
        Package nettyPackage = Bootstrap.class.getPackage();
        String title = nettyPackage.getImplementationTitle();
        String version = nettyPackage.getImplementationVersion();
        String logBanner = "\n\n"
                + "* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\n"
                + "*                                                                                   *\n"
                + "*                                                                                   *\n"
                + "*                   Netty Http Server with Continuable started on                   *\n"
                + "*                  port:{} ,protocol:{},nProcessor:{},timeout:{}         *\n"
                + "*                      Running with {} {}                    *\n"
                + "*                                                                                   *\n"
                + "*                                                                                   *\n"
                + "* * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * * *\n";
        log.info(logBanner, port, channelInitializer.getProtocol(), channelInitializer.getnProcessors(),
                channelInitializer.getTimeOut(), title, version);
    }

    protected EventLoopGroup initEventLoopGroup(int nThreads) {
        if (Epoll.isAvailable()) {
            return new EpollEventLoopGroup(nThreads);
        } else if (KQueue.isAvailable()) {
            return new KQueueEventLoopGroup(nThreads);
        } else {
            return new NioEventLoopGroup(nThreads);
        }
    }

    protected ServerBootstrap initChannel() {
        ServerBootstrap serverBootstrap = new ServerBootstrap().group(boss, worker);
        if (Epoll.isAvailable()) {
            return serverBootstrap.channel(EpollServerSocketChannel.class);
        } else if (KQueue.isAvailable()) {
            return serverBootstrap.channel(KQueueServerSocketChannel.class);
        } else {
            return serverBootstrap.channel(NioServerSocketChannel.class);
        }
    }

    public int getPort() {
        return port;
    }

    public void close() {
        try {
            if (channel != null) {
                channel.closeFuture().sync();
            }
        } catch (InterruptedException e) {
            log.error("server close error", e);
        } finally {
            shutdownGracefully();
        }
    }

    private void shutdownGracefully() {
        if (boss != null) {
            boss.shutdownGracefully().syncUninterruptibly();
        }
        if (worker != null) {
            worker.shutdownGracefully().syncUninterruptibly();
        }
        if (Processor != null) {
            Processor.shutdownGracefully().syncUninterruptibly();
        }
    }

    protected ThreadFactory bossThreadFactory(String name) {
        return new DefaultThreadFactory(name, Thread.MAX_PRIORITY);
    }

    protected ThreadFactory workerThreadFactory(String name) {
        return new DefaultThreadFactory(name, Thread.MAX_PRIORITY);
    }
}
